package com.exercise.rpc.client;/**
 * Created by kete on 2018-10-29.
 */

import com.exercise.rpc.common.decoder.RpcDecoder;
import com.exercise.rpc.common.encoder.RpcEncoder;
import com.exercise.rpc.common.model.RpcRequest;
import com.exercise.rpc.common.model.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 作者：kete
 * 创建时间：2018-10-29 9:55
 * 功能描述：
 * 版本：
 */
public class ConnectManage {
    private static final Logger logger = LoggerFactory.getLogger(ConnectManage.class);

    private final static ConnectManage me = new ConnectManage();

    private EventLoopGroup workGrop = new NioEventLoopGroup(4);

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 16,
            600L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));

    private List<ClientHandler> connectedHandlers = new CopyOnWriteArrayList<>();

    private Map<InetSocketAddress,ClientHandler> connectedServerNodes = new ConcurrentHashMap<>();

    private ReentrantLock lock = new ReentrantLock();

    private Condition condition = lock.newCondition();

    private long connectTimeoutMillis = 6000;

    private AtomicInteger roundRobin = new AtomicInteger(0);

    private volatile boolean isRuning = true;

    private ConnectManage() {

    }

    public static ConnectManage me(){
        return me;
    }

    /**
     * 更新服务连接
     * @param allServerAddress
     */
    public void updateConnectedServer(List<String> allServerAddress){
        if(allServerAddress!=null){
            if(allServerAddress.size()>0){

                HashSet<InetSocketAddress> newAllServerNodeSet = new HashSet<InetSocketAddress>();

                allServerAddress.forEach(serverAddress->{
                    String host = serverAddress.split(":")[0];

                    int port = Integer.parseInt(serverAddress.split(":")[1]);

                    final InetSocketAddress remotePeer = new InetSocketAddress(host,port);

                    newAllServerNodeSet.add(remotePeer);
                });

                //不存在,则建立连接
                newAllServerNodeSet.forEach(serverNodeAddress->{
                    if(!connectedServerNodes.keySet().contains(serverNodeAddress)){
                        connectedNode(serverNodeAddress);
                    }
                });

                // Close and remove invalid server nodes
                connectedHandlers.forEach(handler->{
                    SocketAddress remotePeer = handler.getRemotePeer();
                    if(!newAllServerNodeSet.contains(remotePeer)){
                        logger.info("Remove invalid server node " + remotePeer);
                        if(connectedServerNodes.get(remotePeer)!=null)
                            connectedServerNodes.get(remotePeer).close();
                        connectedServerNodes.remove(remotePeer);
                        connectedHandlers.remove(handler);
                    }
                });
            }
        }else{
            connectedHandlers.forEach(handler->{
                SocketAddress remotePeer = handler.getRemotePeer();
                if(connectedServerNodes.get(remotePeer)!=null)
                    connectedServerNodes.get(remotePeer).close();
                connectedServerNodes.remove(remotePeer);
            });

            connectedHandlers.clear();
        }

    }

    /**
     * 连接服务节点
     * @param address
     */
    private void connectedNode(final InetSocketAddress address){
        executor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    Bootstrap bootstrap = new Bootstrap();
                    bootstrap.group(workGrop)
                            .channel(NioSocketChannel.class)
                            .handler(new InitClientHandler());

                    ChannelFuture future = bootstrap.connect(address).sync();
                    future.addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(final ChannelFuture future) throws Exception {
                            if(future.isSuccess()){
                                logger.debug("Successfully connect to remote server. remote peer = " + address);
                                ClientHandler handler = future.channel().pipeline().get(ClientHandler.class);
                                addHandler(handler);
                            }
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 重新连接
     * @param handler
     * @param address
     */
    private void reconnected(final ClientHandler handler, final SocketAddress address){
        if (handler != null) {
            connectedHandlers.remove(handler);
            connectedServerNodes.remove(handler.getRemotePeer());
        }
        connectedNode((InetSocketAddress) address);
    }


    private void addHandler(ClientHandler handler){
        connectedHandlers.add(handler);
        InetSocketAddress address = (InetSocketAddress) handler.getRemotePeer();
        connectedServerNodes.put(address,handler);

        signalAvailableHandler();

    }

    private void signalAvailableHandler() {
        lock.lock();
        try {
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    private boolean waitingForHandler() throws InterruptedException{
        lock.lock();
        try {
            return condition.await(this.connectTimeoutMillis,TimeUnit.MILLISECONDS);
        } finally {
            lock.unlock();
        }

    }

    /**
     * 返回可用连接
     * @return
     */
    public ClientHandler chooseHandler(){
        int size = connectedHandlers.size();

        //connect not initialize
        while (isRuning&&size<=0){
            try {
                boolean available = waitingForHandler();
                if(available){
                    size = connectedHandlers.size();
                }
            } catch (InterruptedException e) {
                logger.error("Waiting for available node is interrupted! ", e);
                throw new RuntimeException("Can't connect any servers!", e);
            }

        }

        int index = (roundRobin.getAndAdd(1) + size) % size;
        return connectedHandlers.get(index);
    }

    public void stop(){

        isRuning = false;
        connectedHandlers.forEach(handler -> {
            handler.close();
        });

        signalAvailableHandler();
        executor.shutdown();
        workGrop.shutdownGracefully();
    }

    private class InitClientHandler extends ChannelInitializer<SocketChannel>{
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline cp = ch.pipeline();
            cp.addLast(new RpcEncoder(RpcRequest.class));
            cp.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 0));
            cp.addLast(new RpcDecoder(RpcResponse.class));
            cp.addLast(new ClientHandler());
        }
    }
}
